package com.darrenchan.spark

import java.sql.{DriverManager, PreparedStatement}

import org.apache.spark.SparkConf
import org.apache.spark.streaming.{Seconds, StreamingContext}

/**
  * 使用Spark Streaming完成词频统计，并写入MySQL
  */
object ForeachRDDApp {

  def main(args: Array[String]): Unit = {


    val sparkConf = new SparkConf().setAppName("ForeachRDDApp").setMaster("local[2]")
    val ssc = new StreamingContext(sparkConf, Seconds(5))

    // 如果使用了stateful的算子，必须要设置checkpoint
    // 在生产环境中，建议大家把checkpoint设置到HDFS的某个文件夹中
    //    ssc.checkpoint(".")

    val lines = ssc.socketTextStream("hadoop000", 6789)

    val result = lines.flatMap(_.split(" ")).map((_, 1)).reduceByKey(_+_)

    result.print() //此处仅仅是将统计结果输出到控制台

    //将结果写入到MySQL
    result.foreachRDD(rdd => {
      rdd.foreachPartition(partitionOfRecords => {
          val conn = createConnection()
          var sql: String = null
          partitionOfRecords.foreach(record => {

            sql = "select * from wordcount where word = ?"
            var pstmt = conn.prepareStatement(sql)
            val word = record._1
            pstmt.setString(1, word)

            val resultSet = pstmt.executeQuery()
            var ww: WordCount = null
            while(resultSet.next()) {
              val word = resultSet.getString("word")
              val wordcount = resultSet.getInt("wordcount")
              ww = new WordCount(word, wordcount)
            }

            var count: Int = 0
            if (ww != null) {
              count = ww.wordcount + record._2
              sql = "update wordcount set wordcount = ? where word = ?"
              pstmt = conn.prepareStatement(sql)
              pstmt.setInt(1, count)
              pstmt.setString(2, ww.word)
            } else {
              count = record._2
              sql = "insert into wordcount(word, wordcount) values (?, ?)"
              pstmt = conn.prepareStatement(sql)
              pstmt.setString(1, word)
              pstmt.setInt(2, count)
            }

            pstmt.executeUpdate()
            pstmt.close()

          })
          conn.close()

      })
    })


    ssc.start()
    ssc.awaitTermination()
  }

  /**
    * 获取mysql的连接
    *
    * @return
    */
  def createConnection() = {
    Class.forName("com.mysql.jdbc.Driver")
    DriverManager.getConnection("jdbc:mysql://localhost:3306/imooc_spark", "root", "1248")
  }

}

class WordCount(val word: String, val wordcount: Int) {}
